package engine

import (
	"context"
	"fmt"
	"io"
	"regexp"
	"strings"
	"sync"
	"time"

	"gitee.com/zfd81/dbrs/sql"
	"gitee.com/zfd81/dbrs/sql/annotation"
	"gitee.com/zfd81/dbrs/sql/parser"

	"gitee.com/zfd81/dbrs/config"

	"google.golang.org/grpc"
	"google.golang.org/grpc/balancer/roundrobin"
	"google.golang.org/grpc/resolver"
	"google.golang.org/grpc/resolver/manual"

	"github.com/spf13/cast"

	pb "gitee.com/zfd81/dbrs/proto/dbrspb"

	"gitee.com/zfd81/dbrs/internal/token"

	"vitess.io/vitess/go/sqltypes"
	querypb "vitess.io/vitess/go/vt/proto/query"
)

const (
	Token_Secret = "1q2w3e"
	Token_Minute = 10 * 60
)

func getToken(ctx *Context) (string, error) {
	var sign string
	a := ctx.GetAnnotation(annotation.Type_Sign)
	if len(a) < 1 {
		val, err := ctx.GetUserVariable(ctx, "user_sign")
		if err != nil {
			return "", err
		}
		if val != nil {
			sign = cast.ToString(val)
		} else {
			return "", fmt.Errorf("user_sign not found in query: %s", strings.TrimSpace(ctx.Query()))
		}
	} else {
		sign = a[0].String()
	}
	claims := token.BrokerClaims{
		Tenant: ctx.Client().User,
		Sign:   sign,
		User:   ctx.Client().User,
	}
	return token.Create(claims, Token_Secret, Token_Minute)
}

type DatabrokerProvider struct {
	name   string
	typ    Type
	client pb.BrokerClient
	mu     *sync.RWMutex
}

func (p *DatabrokerProvider) Name() string {
	return p.name
}

func (p *DatabrokerProvider) Type() Type {
	return p.typ
}

func (p *DatabrokerProvider) Insert(ctx *Context, stmt *parser.InsertStatement) (*sqltypes.Result, error) {
	return nil, fmt.Errorf("Insert statement is not currently supported")
}

func (p *DatabrokerProvider) Delete(ctx *Context, stmt *parser.DeleteStatement) (*sqltypes.Result, error) {
	return nil, fmt.Errorf("Delete statement is not currently supported")
}

func (p *DatabrokerProvider) Update(ctx *Context, stmt *parser.UpdateStatement) (*sqltypes.Result, error) {
	return nil, fmt.Errorf("Update statement is not currently supported")
}

func (p *DatabrokerProvider) Select(ctx *Context, bindVars map[string]*querypb.BindVariable, callback func(*sqltypes.Result) error, stmt *parser.SelectStatement) error {
	token, err := getToken(ctx)
	if err != nil {
		return err
	}

	vers := ctx.GetAnnotation(annotation.Type_Version)

	request := &pb.StreamRowsRequest{
		Token: token,
		Query: ctx.Sql(),
	}
	for _, ver := range vers {
		strs := strings.Split(ver.String(), ",")
		if len(strs) != 2 {
			return fmt.Errorf("Version annotation format error: @ver(%s)", ver.String())
		}
		request.Versions = append(request.Versions, &pb.Version{
			Name: strs[0],
			Num:  strs[1],
		})
	}

	resp, err := p.client.StreamRows(context.Background(), request)
	if err != nil {
		return err
	}
	for {
		entry, err := resp.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			return err
			break
		}
		fs := entry.Fields
		rs := entry.Rows
		s := gfs2Schema(fs)
		r := &sqltypes.Result{Fields: s.ToFields()}
		for _, grow := range rs {
			row := make([]sqltypes.Value, len(s))
			for i, col := range grow.Columns {
				row[i] = sqltypes.MakeTrusted(s[i].Type, []byte(col))
			}
			r.Rows = append(r.Rows, row)
		}
		r.RowsAffected = uint64(len(r.Rows))
		callback(r)
	}
	return nil
}

func gfs2Schema(fs []*pb.Field) sql.Schema {
	schema := make(sql.Schema, len(fs))
	for i, f := range fs {
		schema[i] = Field2Column(f)
	}
	return schema
}

func Field2Column(f *pb.Field) *sql.Column {
	return &sql.Column{
		Name: f.Name,
		Type: querypb.Type(f.Type),
	}
}

func (p *DatabrokerProvider) ShowTables(ctx *Context, stmt *parser.ShowStatement) (*sqltypes.Result, error) {
	token, err := getToken(ctx)
	if err != nil {
		return nil, err
	}
	request := &pb.GetSchemaRequest{
		Token: token,
	}
	resp, err := p.client.GetSchema(context.Background(), request)
	if err != nil {
		return nil, err
	}

	r := &sqltypes.Result{}
	schema := stmt.Node().Schema()
	fields := schema.ToFields()
	r.Fields = fields
	reg := "^" + strings.ReplaceAll(stmt.Filter, "%", "\\w*") + "$"
	for _, t := range resp.TableDefinitions {
		matched, _ := regexp.MatchString(reg, t.Name)
		if stmt.Filter == "" || matched {
			row := schema.CreateRow()
			row.Append(t.Name)
			if stmt.Full {
				row.Append(t.Type)
			}
			row.Append(t.Comment)
			row.Append(t.Version)
			r.Rows = append(r.Rows, row.ToSQL())
		}
	}
	r.RowsAffected = uint64(len(r.Rows))
	return r, nil
}

func (p *DatabrokerProvider) ShowCharset(ctx *Context, stmt *parser.ShowStatement) (*sqltypes.Result, error) {
	r := &sqltypes.Result{}
	schema := stmt.Node().Schema()
	fields := schema.ToFields()
	r.Fields = fields
	r.RowsAffected = 1
	row := schema.CreateRow()
	row.Append("utf8mb4", "UTF-8 Unicode", "utf8mb4_0900_ai_ci", 4)
	r.Rows = append(r.Rows, row.ToSQL())
	return r, nil
}
func (p *DatabrokerProvider) ShowCollation(ctx *Context, stmt *parser.ShowStatement) (*sqltypes.Result, error) {
	r := &sqltypes.Result{}
	schema := stmt.Node().Schema()
	fields := schema.ToFields()
	r.Fields = fields
	r.RowsAffected = 1
	row := schema.CreateRow()
	row.Append("latin1_german1_ci", "latin1", 5, "No", "Yes", 1, "PAD SPACE")
	r.Rows = append(r.Rows, row.ToSQL())
	return r, nil
}

func (p *DatabrokerProvider) DescTable(ctx *Context, stmt *parser.DescStatement) (*sqltypes.Result, error) {
	token, err := getToken(ctx)
	if err != nil {
		return nil, err
	}
	request := &pb.GetTableRequest{
		Token: token,
		Name:  stmt.Tbl(),
	}
	resp, err := p.client.GetTable(context.Background(), request)
	if err != nil {
		return nil, err
	}
	r := &sqltypes.Result{}
	schema := stmt.Node().Schema()
	fields := schema.ToFields()
	r.Fields = fields
	for _, col := range resp.Columns {
		row := schema.CreateRow()
		row.Append(col.Name)
		row.Append(col.Type)
		row.Append(col.Nullable)
		row.Append("")
		row.Append(col.Default)
		row.Append("")
		row.Append(col.Comment)
		r.Rows = append(r.Rows, row.ToSQL())
	}
	r.RowsAffected = uint64(len(r.Rows))
	return r, nil
}

func (p *DatabrokerProvider) AlterTable(ctx *Context, stmt *parser.AlterTableStatement) (*sqltypes.Result, error) {
	return nil, fmt.Errorf("Alter table statement is not currently supported")
}

func (p *DatabrokerProvider) DropTable(ctx *Context, stmt *parser.DropTableStatement) (*sqltypes.Result, error) {
	return nil, fmt.Errorf("Drop table statement is not currently supported")
}

func (p *DatabrokerProvider) Client() pb.BrokerClient {
	return p.client
}

func NewDatabrokerProvider(db config.Database) (*DatabrokerProvider, error) {
	var addresses []resolver.Address
	for _, endpoint := range db.Endpoints {
		addresses = append(addresses, resolver.Address{Addr: endpoint})
	}

	r := manual.NewBuilderWithScheme("whatever")
	r.InitialState(resolver.State{Addresses: addresses})
	conn, err := grpc.Dial(
		r.Scheme()+":///grpcs.server",
		grpc.WithInsecure(),
		grpc.WithResolvers(r),
		grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s", "RetryPolicy": {"MaxAttempts":2, "InitialBackoff": "0.1s", "MaxBackoff": "1s", "BackoffMultiplier": 2.0, "RetryableStatusCodes": ["UNAVAILABLE"]}}`, roundrobin.Name)),
		grpc.WithBlock(),
		grpc.WithTimeout(3*time.Second))
	if err != nil {
		return nil, fmt.Errorf("did not connect broker: %v", err)
	}

	return &DatabrokerProvider{
		name:   db.Name,
		typ:    Type_Databroker,
		client: pb.NewBrokerClient(conn),
		mu:     &sync.RWMutex{},
	}, nil
}
